#!/bin/bash
set -e

clickhouse client -n <<-EOSQL
    CREATE TABLE IF NOT EXISTS flows
    (
        TimeReceived DateTime,


        SamplingRate UInt64,
        SamplerAddress FixedString(16),

        SrcAddr FixedString(16),
        DstAddr FixedString(16),


        EType UInt32,

        SrcPort UInt32,
        DstPort UInt32,

        Out Bool,

        Bytes UInt64,
        Packets UInt64
    ) ENGINE = Kafka()
    SETTINGS
        kafka_broker_list = 'netopskafka:9092',
        kafka_topic_list = 'flows',
        kafka_group_name = 'clickhouse',
        kafka_format = 'Protobuf',
        kafka_schema = 'flow.proto:FlowMessage';

    CREATE TABLE IF NOT EXISTS flows_raw
    (
        Date Date,
        TimeReceived DateTime,

        SamplingRate UInt64,
        SamplerAddress FixedString(16),

        SrcAddr FixedString(16),
        DstAddr FixedString(16),


        EType UInt32,

        SrcPort UInt32,
        DstPort UInt32,

        Out Bool,

        Bytes UInt64,
        Packets UInt64
    ) ENGINE = MergeTree()
    PARTITION BY toYYYYMMDD(Date)
    ORDER BY (SamplerAddress,Out)
    PRIMARY KEY (SamplerAddress,Out)
    TTL Date+INTERVAL 2 DAY;



    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_raw_view TO flows_raw
    AS SELECT
       toDate(TimeReceived) AS Date,
       *
       FROM flows;




CREATE TABLE IF NOT EXISTS flows_1m
    (
        Timeslot DateTime,
        Out Bool,
        SamplerAddress FixedString(16),

        Bytes UInt64

    ) ENGINE = SummingMergeTree()
    PARTITION BY toYYYYMM(Timeslot)
    ORDER BY (Timeslot,Out,SamplerAddress)
    PRIMARY KEY (Timeslot,Out,SamplerAddress);


    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_1m_view TO flows_1m
    AS
        SELECT
            toStartOfMinute(TimeReceived) AS Timeslot,
			Out,
            IPv4NumToString(reinterpretAsUInt32(substring(reverse(SamplerAddress), 13,4))) as SamplerAddress,
            sum((Bytes * SamplingRate) / 8) AS Bytes

        FROM flows_raw
        GROUP BY Timeslot,Out,SamplerAddress;


CREATE TABLE IF NOT EXISTS flows_in_1m
    (
        Timeslot DateTime,
        SrcAddr FixedString(16),
        DstAddr FixedString(16),
        SamplerAddress FixedString(16),
        Bytes UInt64

    ) ENGINE = SummingMergeTree()
    PARTITION BY toYYYYMM(Timeslot)
    ORDER BY (SamplerAddress,SrcAddr,DstAddr,Timeslot)
    PRIMARY KEY (SamplerAddress,SrcAddr,DstAddr,Timeslot);


    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_in_1m_view TO flows_in_1m
    AS
        SELECT
            toStartOfMinute(TimeReceived) AS Timeslot,
			SrcAddr,
			DstAddr,
            IPv4NumToString(reinterpretAsUInt32(substring(reverse(SamplerAddress), 13,4))) as SamplerAddress,
            sum((Bytes * SamplingRate) / 8) AS Bytes

        FROM flows_raw
        where Out=false
        GROUP BY Timeslot,SrcAddr,DstAddr,SamplerAddress;

CREATE TABLE IF NOT EXISTS flows_out_1m
    (
        Timeslot DateTime,
        SrcAddr FixedString(16),
        DstAddr FixedString(16),
        SamplerAddress FixedString(16),

        Bytes UInt64

    ) ENGINE = SummingMergeTree()
    PARTITION BY toYYYYMM(Timeslot)
    ORDER BY (SamplerAddress,SrcAddr,DstAddr,Timeslot)
    PRIMARY KEY (SamplerAddress,SrcAddr,DstAddr,Timeslot);



    CREATE MATERIALIZED VIEW IF NOT EXISTS flows_out_1m_view TO flows_out_1m
    AS
        SELECT
            toStartOfMinute(TimeReceived) AS Timeslot,
			SrcAddr,
			DstAddr,
            IPv4NumToString(reinterpretAsUInt32(substring(reverse(SamplerAddress), 13,4))) as SamplerAddress,
            sum((Bytes * SamplingRate) / 8) AS Bytes

        FROM flows_raw
        where Out=true
        GROUP BY Timeslot,SrcAddr,DstAddr,SamplerAddress;

EOSQL
